package cn.darksoul3.spark.operator

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object LeftOuterJoin {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("group by key").setMaster("local[*]")

    val sc = new SparkContext(conf)

    val rdd1: RDD[(String, Int)] = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2)), numSlices = 2)
    val rdd2: RDD[(String, Int)] = sc.parallelize(List(("tom", 1), ("jerry", 2), ("tom", 3), ("penny", 2)), numSlices = 2)
    val rdd3: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)
    val leftOuterJoin: RDD[(String, (Int, Option[Int]))] = rdd3.flatMapValues { pair =>
      if (pair._2.isEmpty) {
        pair._1.iterator.map(v => (v, None))
      } else {
        for (x <- pair._1.iterator; y <- pair._2.iterator) yield (x, Some(y))
      }
    }
    leftOuterJoin.saveAsTextFile("leftOuterJoin-out")
  }
}
